package atlub.tweet_crawler;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.Twitter;

public class CrawlerThreadController {
    ICredentials c;
    List<Long> userIds;
    List<CrawlerThread> threads;
    MonitorThread mthread;
    BlockingQueue<Long> userPool;
    int totalUsers = -1;
    
    public CrawlerThreadController() throws Exception{
	c = new CredentialsRoundrobin();
	threads = new ArrayList<CrawlerThread>(c.getCredentialCount());
	userIds = CrawlerThread.getAllUserIds();
	userPool = new LinkedBlockingQueue<Long>();
	for(Long userId:userIds){userPool.add(userId);} 
	totalUsers = userPool.size();
    }
    
    public void startJobs(){
	for(int i = 0;i < c.getCredentialCount();++i){
	    Twitter twitter = new TwitterAgent(c.getCredential()).getTwitter();
	    CrawlerThread thread = new CrawlerThread(this, userPool, twitter);
	    threads.add(thread);
	}
	
	for(int i = 0;i < threads.size();++i){
	    threads.get(i).start();
	}
	mthread = new MonitorThread(userPool);
	mthread.start();
    }
    public void stopJobs(){
	for(int i = 0;i < threads.size();++i){
	    threads.get(i).isFinished = true; //soft interrupt
	}
	mthread.interrupt();
    }
    public void interruptJobs(){
	for(int i = 0;i < threads.size();++i){
	    threads.get(i).interrupt();
	}
	mthread.interrupt();
    }
    
    public static void main(String args[]){
	try{
	    CrawlerThreadController ctc = new CrawlerThreadController();
	    ctc.startJobs();
	}catch(Exception e){e.printStackTrace();}
	
    }
    
    class MonitorThread extends Thread{
	BlockingQueue<Long> userPool;
	public MonitorThread(BlockingQueue<Long> userPool){
	    this.userPool = userPool;
	}
	
	@Override
	public void run() {
	    
	    try{
		while(true){
		    Thread.sleep(3*1000);
		    if(userPool.size()==0){break;}
		    System.out.println("Processed: "+userPool.size()+"/"+totalUsers);
		}
	    }catch(Exception e){e.printStackTrace();}
	}
	
    }
}
